package kn.xianSubway.mq;


import cn.hutool.http.HttpStatus;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.rabbitmq.client.Channel;
import kn.xianSubway.common.Constants;
import kn.xianSubway.common.OkHttpUtils;
import kn.xianSubway.req.ProductMsgBean;
import kn.xianSubway.resp.MatchDistanceResp;
import kn.xianSubway.service.ExternalService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Component
@Slf4j
public class Consumer {


    @Autowired
    private ExternalService externalService;


    @Autowired
    @Qualifier("sjTopicMqRabbitTemplate")
    private RabbitTemplate sjTopicMqRabbitTemplate;




    @Value("${monitor.keyword.saveUrl}")
    private String getKeywordSaveUrl;

    /**
     *  监听pushData服务发送的西安地铁的消息
     *  公共队列消费
     * @param message
     * @param channel
     * @param rabbitMessage
     */
    @RabbitListener(queues = Constants.XIAN_SUBWAY_QUEUE,containerFactory = "xianSubwayFactory")
    public void keywordReceiveMessage0(String message, Channel channel, Message rabbitMessage) {

        try {
            MatchDistanceResp resp = JSON.parseObject(message, MatchDistanceResp.class);
            log.info("收到西安地铁AI调性请求AI的消息，链接：{}", resp.getWebsite());
            // 原调性为负面

            if("2".equals(resp.getTonalState())){
                // 负面推送未研判前群消息
                // sendbefor(resp,"西安地铁-未过AI工作群","xian_monitor_key");
                // 过豆包AI
                String sb = resp.getContent() + resp.getImageContent() + resp.getTitle() +
                        resp.getSource() + resp.getAuthor() + resp.getWebsite();
                long startTime = System.currentTimeMillis();
                Map<String, String> resultMap = externalService.queryNewTonalState(sb, resp.getWebsite(),"bot-20250221114550-r6tfv","bot-20250219170242-clxxl");
                long endTime = System.currentTimeMillis();
                log.info("西安地铁新调性请求耗时:{}ms", (endTime - startTime));
                resp.setAiTonalStateReason(resultMap.getOrDefault("理由","无"));
                if (!resultMap.isEmpty()) {
                    if ("是".equals(resultMap.get("相关"))) {
                        String tonalStateNumber = getTonalState(resultMap.get("调性"));
                        resp.setTonalState(tonalStateNumber);
                        resp.setIsAiTonalState(1);
                        log.info("西安地铁链接：{}相关为是则入库", resp.getWebsite());
                        // 相关为是 入库
                        saveAndSend(Collections.singletonList(resp));
                    } else {
                        log.info("西安地铁链接：{}相关为否则不入库", resp.getWebsite());
                    }
                }else {
                    log.info("西安地铁链接：{}研判调性结果为空用原调性", resp.getWebsite());
                    saveAndSend(Collections.singletonList(resp));
                }
            }else {
                // 原调性为 正面 或者 中性 入库
                log.info("西安地铁链接：{}原调性为正面或者中性入库", resp.getWebsite());
                saveAndSend(Collections.singletonList(resp));
            }
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
        }catch (JSONException jsonException){
            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
            try {
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }

        }catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }


    /**
     *  监听pushData服务发送的沈阳的消息
     *  公共队列消费
     * @param message
     * @param channel
     * @param rabbitMessage
     */
    @RabbitListener(queues = Constants.SHENYANG_MANAGE_QUEUE,containerFactory = "xianSubwayFactory")
    public void keywordReceiveMessage1(String message, Channel channel, Message rabbitMessage) {

        try {
            MatchDistanceResp resp = JSON.parseObject(message, MatchDistanceResp.class);
            log.info("收到沈阳应急管理局AI调性请求AI的消息，链接：{}", resp.getWebsite());
            // 原调性为负面

            if("2".equals(resp.getTonalState())){
                // 负面推送未研判前群消息
                //sendbefor(resp, "沈阳市应急管理局-未过AI工作群","shenyang_monitor_key");
                // 过豆包AI
                String sb = resp.getContent() + resp.getImageContent() + resp.getTitle() +
                        resp.getSource() + resp.getAuthor() + resp.getWebsite();
                long startTime = System.currentTimeMillis();
                Map<String, String> resultMap = externalService.queryNewTonalState(sb, resp.getWebsite(),"bot-20250304104957-twqht","bot-20250304091348-jrcjn");
                long endTime = System.currentTimeMillis();
                log.info("沈阳应急管理局新调性请求耗时:{}ms", (endTime - startTime));
                resp.setAiTonalStateReason(resultMap.getOrDefault("理由","无"));
                if (!resultMap.isEmpty() && resp.getIsShenYang() == 1) {
                    if ("是".equals(resultMap.get("相关"))) {
                        String tonalStateNumber = getTonalState(resultMap.get("调性"));
                        resp.setTonalState(tonalStateNumber);
                        resp.setIsAiTonalState(1);
                        log.info("沈阳应急管理局链接：{}相关为是则入库", resp.getWebsite());
                        // 相关为是 入库
                        saveAndSend(Collections.singletonList(resp));
                    } else {
                        log.info("沈阳应急管理局链接：{}相关为否则不入库", resp.getWebsite());
                    }
                }else {
                    log.info("沈阳应急管理局链接：{}研判调性结果为空用原调性", resp.getWebsite());
                    saveAndSend(Collections.singletonList(resp));
                }
            }else {
                // 原调性为 正面 或者 中性 入库
                log.info("沈阳应急管理局链接：{}原调性为正面或者中性入库", resp.getWebsite());
                saveAndSend(Collections.singletonList(resp));
            }
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
        }catch (JSONException jsonException){
            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
            try {
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }
    private void sendbefor(MatchDistanceResp resp, String groupName,String key) {
        ProductMsgBean productMsgBean = new ProductMsgBean();
        String releaseTime = resp.getReleaseTime();

        productMsgBean.setMsgId(String.valueOf(resp.getDataID()));
        //productMsgBean.setOrganId(Long.valueOf(p.getAgencyId()));
        productMsgBean.setPushChannel(1);
        productMsgBean.setPushGroupName(groupName);
        String text = resp.getContent();
        String abstracts = text.length() >= 300 ? text.substring(0,300) : text;

        String content = "标题:" + resp.getTitle() + "\n" +
                "发布时间:"+ releaseTime + "\n" +
                "发布平台:"+ resp.getSource() + "\n" +
                "发布人:"+ resp.getAuthor() + "\n" +
                "摘要:"+ abstracts + "\n" +
                "链接:"+ resp.getWebsite() + "\n";

        productMsgBean.setContent(content);
        String msg = JSON.toJSONString(productMsgBean);
        sjTopicMqRabbitTemplate.convertAndSend("xian_monitor_exchange",key, msg);
        log.info("发送西安地铁未研判调性之前的消息到队列：{}", msg);
    }

    private void saveAndSend(List<MatchDistanceResp> resps) {
        String req = JSON.toJSONString(resps, SerializerFeature.WriteNullNumberAsZero, SerializerFeature.WriteNullListAsEmpty);
        long start = System.currentTimeMillis();
        if (pullData(req, getKeywordSaveUrl)) {
            log.info("推送数据成功：{}",JSONUtil.toJsonStr(resps.stream().map(MatchDistanceResp::getWebsite).collect(Collectors.toList())));
        }else {
            log.error("推送数据重试次数超过后异常：{}",JSONUtil.toJsonStr(req));
        }
        long end = System.currentTimeMillis();
        log.info("推送数据耗时：{}毫秒",(end - start));
    }


    private boolean pullData(String req, String url){
        // 处理消息逻辑
        int maxRetries = 3; // 最大重试次数
        int attempts = 0; // 当前尝试次数
        boolean success = false;

        while (attempts < maxRetries) {
            try {
                String body = OkHttpUtils.postJson(url, req);
                cn.hutool.json.JSONObject jsonObject = JSONUtil.parseObj(body);
                if (HttpStatus.HTTP_OK == jsonObject.getInt("code")) {
                    success = true; // 请求成功，退出重试
                    log.info("推送成功！！！！！！！！！");
                    return success;
                } else {
                    log.error("推送失败");
                    attempts++;

                    // 增加等待时间，避免频繁请求
                    try {
                        Thread.sleep(1000); // 等待1秒后重试
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt(); // 恢复中断状态
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
                throw new RuntimeException("推送数据异常:"+ e.getMessage());
            }

        }
        return false;
    }

    private String getTonalState(String tonalState) {

        if("正面".equals(tonalState)){
            return "0";
        }else if("中性".equals(tonalState)){
            return "1";
        }else if("负面".equals(tonalState)){
            return "2";
        }else {
            throw new RuntimeException("调性返回异常");
        }
    }
}
